package taskpool

import (
	"context"
	"runtime/debug"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

const (
	Success      = "01"
	Fail         = "02"
	Proc         = "03"
	DispatcherID = 0
)

type TaskItemStat struct {
	ID     int
	Status string
	Err    error
}

type WorkerStat struct {
	ID          int
	SuccessTask int64
	FailTask    int64
	Status      string
	err         error
}

type (
	Dispatcher    = func(context.Context, chan interface{}, *int64) error
	SubWorker     = func(context.Context, interface{}) error
	SubLongWorker = func(context.Context, chan interface{}, *WorkerStat) error
)

var chan2List = func(ec chan error) []error {
	var errList []error
tags:
	for {
		select {
		case e := <-ec:
			errList = append(errList, e)
		default:
			break tags
		}
	}
	return errList
}

func (p *TaskPool) PoolRun(ctx context.Context, conCurrency, timeOut int, d Dispatcher, w SubWorker) error {
	logger := p.logger
	logger.Infoc(ctx, "PoolRun 开始运行;")

	statList, err := p.PoolRunStatus(ctx, conCurrency, timeOut, d, w)
	if err != nil {
		logger.Errorc(ctx, "PoolRunStatus 执行失败:[%s]", err)
		for _, stat := range statList {
			logger.Errorc(ctx, "PoolRunStatus subTask stat[%+v]", stat)
		}
		return err
	}

	logger.Infoc(ctx, "PoolRun 执行完成;")
	return nil
}

// PoolRunStatus 并行执行任务；分发worker.
func (p *TaskPool) PoolRunStatus(ctx context.Context, conCurrency, timeOut int, d Dispatcher, w SubWorker) ([]TaskItemStat, error) {
	var wg sync.WaitGroup
	var workerResult string
	var dispatcherResult string
	var dispatcherCount int64 = 0
	var workerCount int64 = 0
	logger := p.logger

	ctx = logger.ExtendTracerID(ctx)

	logger.Infoc(ctx, "PoolRunStatus 开始运行;")
	begin := time.Now()

	// 最少默认值
	if conCurrency <= 0 {
		conCurrency = defaultPoolSize
	}

	// 并发数默认值
	if conCurrency > int(p.maxWorkers) {
		logger.Infoc(ctx, "PoolRun 入参并发[%d] taskPool 实际并发数[%d] 支持最大并发数[%d]", conCurrency, p.maxWorkers, p.maxWorkers)
		conCurrency = int(p.maxWorkers)
	}

	// 最少等待5秒后，退出
	if timeOut < 5 {
		timeOut = 5
	}

	// worker 分发通道
	taskChan := make(chan interface{})
	// 接收worker 处理结果
	statChan := make(chan *TaskItemStat, conCurrency)
	// 状态统计
	statList := make([]TaskItemStat, 0)
	// error列表
	errorChan := make(chan error, conCurrency+1)

	nCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 回收work callback
	wg.Add(1)
	go func() {
		defer wg.Done()
		callbackFunc(ctx, logger, conCurrency, statChan, statList, errorChan, cancel, &workerResult)
	}()

	// 启动worker 接受task
	for i := 1; i <= conCurrency; i++ {
		wg.Add(1)
		logger.Infoc(nCtx, "开始启动worker [%d];", i)
		// 添加worker 到work pool
		idx := i
		p.AddFunc(func() {
			defer wg.Done()
			workFunc1(nCtx, logger, w, taskChan, idx, timeOut, &workerCount, statChan)
		})
	}

	// 开始启动 dispatcher
	wg.Add(1)
	go func() {
		defer wg.Done()
		dispatcherFunc(nCtx, logger, d, taskChan, cancel, errorChan, &dispatcherCount, &dispatcherResult)
	}()

	// 等待所有子 goroutine 执行结束
	wg.Wait()

	logger.Infoc(ctx, "PoolRunStatus 执行完成; Worker[%s] Dispatcher[%s] 耗时[%v]", workerResult, dispatcherResult, time.Since(begin))
	if workerResult == Fail || dispatcherResult == Fail || len(errorChan) > 0 {
		errList := chan2List(errorChan)
		logger.Errorc(ctx, "Pool任务执行失败, 分发状态[%s] 执行状态[%s] 子任务执行状态列表[%+v] Error list[%+v]", dispatcherResult, workerResult, statList, errList)
		return statList, errors.Errorf("Pool任务执行失败, 分发状态[%s] 执行状态[%s] 子任务执行状态列表[%+v] Error list[%+v]", dispatcherResult, workerResult, statList, errList)
	}

	if dispatcherCount != workerCount {
		logger.Errorc(ctx, "Pool任务执行失败,分发任务[%d]和执行任务[%d]不匹配; 子任务执行状态列表[%+v];", dispatcherCount, workerCount, statList)
		return statList, errors.Errorf("Pool任务执行失败,分发任务[%d]和执行任务[%d]不匹配; 子任务执行状态列表[%+v];", dispatcherCount, workerCount, statList)
	}

	return statList, nil
}

// PoolRunList 并行执行任务列表.
func (p *TaskPool) PoolRunList(ctx context.Context, taskList []interface{}, w SubWorker) ([]TaskItemStat, error) {
	var wg sync.WaitGroup
	var workerResult string
	taskNum := len(taskList)
	logger := p.logger

	ctx = logger.ExtendTracerID(ctx)

	logger.Infoc(ctx, "PoolRunList 开始运行;")
	begin := time.Now()

	// 判断并行任务列表
	if len(taskList) == 0 {
		return nil, errors.Errorf("任务列表为空[%+v]", taskList)
	}

	// 接收worker 处理结果
	statChan := make(chan *TaskItemStat, taskNum)
	statList := make([]TaskItemStat, taskNum)

	errorChan := make(chan error, taskNum)

	nCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 管理worker callback
	wg.Add(1)
	go func() {
		defer wg.Done()
		callbackFunc(ctx, logger, taskNum, statChan, statList, errorChan, cancel, &workerResult)
	}()

	// 按照task分配任务
	for i := 0; i < taskNum; i++ {
		logger.Infoc(nCtx, "分配任务 [%d];", i)
		// 添加worker 到work pool
		wg.Add(1)
		idx := i
		p.AddFunc(func() {
			defer wg.Done()
			workFunc(nCtx, logger, w, taskList[idx], idx, statChan)
		})
	}
	// 等待所有子 goroutine 执行结束
	wg.Wait()
	logger.Infoc(ctx, "PoolRunList 执行完成; Worker[%s]  耗时[%v]", workerResult, time.Since(begin))

	if workerResult == Fail || len(errorChan) > 0 {
		errList := chan2List(errorChan)
		logger.Errorc(ctx, "PoolRunList 执行失败; Worker[%s] 子任务执行状态[%+v] errList[%+v] 耗时[%v]", workerResult, statList, errList, time.Since(begin))
		return statList, errors.Errorf("PoolRunList error[%+v]", errList)
	}
	return statList, nil
}

var dispatcherFunc = func(ctx context.Context, logger log.Logger, d Dispatcher, taskChan chan interface{}, cancel context.CancelFunc, errorChan chan error, dispatcherCount *int64, dispatcherResult *string) {
	var err error
	dCtx := logger.SetFields(ctx, log.Field{
		"dis_id": DispatcherID,
		"task":   "dispatcher",
	})
	*dispatcherResult = Success

	// 分发 task 到 worker
	err = d(dCtx, taskChan, dispatcherCount)
	if err != nil {
		*dispatcherResult = Fail
		errorChan <- err
		logger.Errorc(dCtx, "dispatcher 任务分发处理失败[%s], 异常退出；", err)

		cancel()
		return
	}
	logger.Infoc(dCtx, "dispatcher 任务分发完成, 等待子任务执行结果;")
	// 任务已完成分发，结束相关子任务以及任务分发任务
	close(taskChan)

	return
}

var callbackFunc = func(ctx context.Context, logger log.Logger, taskTotal int, statChan chan *TaskItemStat, statList []TaskItemStat, errorChan chan error, cancel context.CancelFunc, workerResult *string) {
	mCtx := logger.SetFields(ctx, log.Field{
		"task": "callback",
	})

	taskCnt := 0
	taskFail := 0
	for {
		if taskCnt == taskTotal {
			logger.Infoc(mCtx, "callbackFunc: 所有worker均已执行完成:[%d]累计任务[%d]失败数[%d]", taskCnt, taskTotal, taskFail)

			*workerResult = Success
			// 任务已结束
			if taskFail > 0 {
				*workerResult = Fail
			}
			logger.Infoc(mCtx, "callbackFunc: pool分发[%d] worker均已执行结果[%s], callbackFunc 退出；", taskTotal, *workerResult)
			break
		}

		logger.Infoc(mCtx, "callbackFunc: 开始接收 worker 执行结果,总计[%d] 已回收[%d]:", taskTotal, taskCnt)

		taskStat := <-statChan
		taskCnt++
		statList = append(statList, *taskStat)

		logger.Infoc(mCtx, "callbackFunc: worker 汇总[%d] 已完成[%d] 失败[%d] 运行中[%d]；执行结果：[%d][%s][%v];", taskTotal, taskCnt, taskFail, taskTotal-taskCnt,
			taskStat.ID, taskStat.Status, taskStat.Err)
		switch taskStat.Status {
		case Success:
			logger.Infoc(mCtx, "callbackFunc: worker ID[%d] 状态[%s]已成功; 运行中[%d]", taskStat.ID, taskStat.Status, taskTotal-taskCnt)
		case Fail:
			taskFail++
			errorChan <- taskStat.Err
			logger.Errorc(mCtx, "callbackFunc: worker ID[%d] 状态[%s]执行失败; 运行中[%d] 失败[%d] [%s]", taskStat.ID, taskStat.Status, taskTotal-taskCnt, taskFail, taskStat.Err)
			// 结束其他work
			cancel()
		}
	}
}

var workFunc = func(ctx context.Context, logger log.Logger, w SubWorker, task interface{}, idx int, statChan chan *TaskItemStat) {
	var err error
	cCtx := logger.ExtendTracerID(ctx)
	cCtx = logger.SetFields(cCtx, log.Field{
		"work_id": idx,
		"task":    "worker",
	})

	defer func() {
		if rErr := recover(); rErr != nil {
			logger.Errorc(cCtx, "workFunc: worker Panic[%s][%s]", rErr, string(debug.Stack()))
			statChan <- &TaskItemStat{
				ID:     idx,
				Status: Fail,
				Err:    errors.Errorf("worker Panic[%s]", rErr),
			}
		}
	}()

	logger.Infoc(cCtx, "workFunc: worker 接收任务[%d];", idx)

	select {
	case <-cCtx.Done(): // 任务退出信号
		statChan <- &TaskItemStat{
			ID:     idx,
			Status: Fail,
			Err:    cCtx.Err(),
		}
		logger.Errorc(cCtx, "workFunc: worker [%d]收到退出信号，不再继续执行", idx)
	default:
		logger.Infoc(cCtx, "workFunc: worker [%d]开始执行任务:", idx)
		err = w(cCtx, task)
		if err != nil {
			statChan <- &TaskItemStat{
				ID:     idx,
				Status: Fail,
				Err:    err,
			}
			logger.Errorc(cCtx, "workFunc: worker [%d]执行任务[%+v]失败[%v],异常退出;", idx, task, err)
			return
		}
		statChan <- &TaskItemStat{
			ID:     idx,
			Status: Success,
		}
		logger.Infoc(cCtx, "workFunc: worker [%d]任务执行成功:", idx)
	}
}

var workFunc1 = func(ctx context.Context, logger log.Logger, w SubWorker, taskChan chan interface{}, idx, timeOut int, workerCount *int64, statChan chan *TaskItemStat) {
	var err error
	begin := time.Now()

	cCtx := logger.SetFields(ctx, log.Field{
		"work_id": idx,
		"task":    "worker",
	})

	defer func() {
		if rErr := recover(); rErr != nil {
			logger.Errorc(cCtx, "worker Panic[%s][%s]", rErr, string(debug.Stack()))
			statChan <- &TaskItemStat{
				ID:     idx,
				Status: Fail,
				Err:    errors.Errorf("worker Panic[%s]", rErr),
			}
		}
	}()

	logger.Infoc(cCtx, "workFunc1: worker[%d] 等待接收任务;", idx)

tag:
	for {
		select {
		case t, ok := <-taskChan:
			if !ok {
				logger.Infoc(cCtx, "workFunc1: worker [%d]读取任务通道结束, 任务通道已关闭,耗时[%v]", idx, time.Since(begin))
				statChan <- &TaskItemStat{
					ID:     idx,
					Status: Success,
				}
				break tag
			}
			wCtx := logger.ExtendTracerID(cCtx)
			logger.Debugc(wCtx, "workFunc1: worker [%d]开始执行任务:", idx)
			err = w(wCtx, t)
			if err != nil {
				statChan <- &TaskItemStat{
					ID:     idx,
					Status: Fail,
					Err:    err,
				}
				logger.Errorc(wCtx, "workFunc1: worker [%d]执行任务[%+v]失败[%v],异常退出;", idx, t, err)
				return
			}
			logger.Debugc(wCtx, "workFunc1: worker [%d]任务执行成功:", idx)
			// 统计执行成功task 数目
			atomic.AddInt64(workerCount, 1)
		case <-cCtx.Done(): // 任务退出信号
			statChan <- &TaskItemStat{
				ID:     idx,
				Status: Fail,
				Err:    cCtx.Err(),
			}
			logger.Infoc(cCtx, "workFunc1: worker [%d]收到退出信号，不再继续执行", idx)
			break tag
		case n := <-time.After(time.Duration(timeOut) * time.Second):
			statChan <- &TaskItemStat{
				ID:     idx,
				Status: Success,
			}
			logger.Infoc(cCtx, "workFunc1: worker [%d]截止到[%v]空闲无任务；正常退出; 耗时[%v]", idx, n, time.Since(begin))
			break tag
		}
	}
}

var workerRun = func(ctx context.Context, idx int, logger log.Logger, taskChan chan interface{}, worker SubLongWorker) (stat *WorkerStat, err error) {
	begin := time.Now()

	cCtx := logger.SetFields(ctx, log.Field{
		"work_id": idx,
		"task":    "worker",
	})

	stat = &WorkerStat{
		ID:          idx,
		SuccessTask: 0,
		FailTask:    0,
		Status:      Proc,
		err:         nil,
	}

	// panic
	defer func() {
		if rErr := recover(); rErr != nil {
			logger.Errorc(cCtx, "worker[%d] Panic[%s][%s]", idx, rErr, string(debug.Stack()))
			stat.Status = Fail
			stat.err = errors.Errorf("worker Panic [%v]", rErr)
			err = errors.Errorf("worker Panic[%v]", rErr)
			return
		}
	}()

	logger.Infoc(cCtx, "workerRun: worker[%d] 开始启动任务执行服务;", idx)
	wCtx := logger.ExtendTracerID(cCtx)

	// worker 需判断taskChan执行结果
	err = worker(wCtx, taskChan, stat)
	if err != nil {
		stat.Status = Fail
		stat.err = err
		logger.Errorc(wCtx, "workerRun: worker[%d] 任务执行失败[%s] 累计执行结果[%+v]", idx, err, stat)
		return stat, err
	}

	logger.Infoc(wCtx, "workerRun: worker[%d] 执行结果[%+v] 累计耗时[%v];", idx, stat, time.Since(begin))
	return stat, nil
}

// WorkerExample workTask 示例
var WorkerExample = func(ctx context.Context, logger log.Logger, taskChan chan interface{}, stat *WorkerStat) error {
	begin := time.Now()
	var err error

	logger.Infoc(ctx, "worker[%d] 开始执行[%v];", stat.ID, time.Since(begin))
	stat.Status = Proc

tag:
	for {
		select {
		case t, ok := <-taskChan:
			if !ok {
				logger.Infoc(ctx, "worker [%d]读取任务通道结束, 任务通道已关闭,耗时[%v]", stat.ID, time.Since(begin))
				stat.Status = Success
				break tag
			}
			logger.Infoc(ctx, "worker [%d]开始执行任务:", stat.ID)
			// 执行业务逻辑
			err = func() error {
				return nil
			}()
			if err != nil {
				stat.FailTask++
				stat.Status = Fail
				logger.Errorc(ctx, "worker [%d]执行任务[%+v]失败[%v],异常退出;", stat.ID, t, err)
				return err
			}
			// 统计执行成功task 数目
			stat.SuccessTask++
			logger.Debugc(ctx, "worker [%d]任务执行成功:", stat.ID)
		case <-ctx.Done(): // 任务退出信号
			logger.Infoc(ctx, "worker [%d]收到退出信号，不再继续执行", stat.ID)
			break tag
		}
	}

	logger.Infoc(ctx, "worker[%d] 累计执行结果[%v] 调用完成;", stat.ID, stat)
	return nil
}

var workerRunCallback = func(ctx context.Context, logger log.Logger, conCurrency int, statChan chan *WorkerStat, errChan chan error, cancel context.CancelFunc) ([]WorkerStat, string) {
	mCtx := logger.SetFields(ctx, log.Field{
		"task": "workerRunCallback",
	})
	var result string
	okCnt := 0
	failCnt := 0
	statList := make([]WorkerStat, 0)
	for {
		if okCnt == conCurrency {
			logger.Infoc(mCtx, "workerRunCallback: 所有worker均已执行完成:[%d]失败数[%d]", okCnt, failCnt)

			result = Success
			// 任务已结束
			if failCnt > 0 {
				result = Fail
			}
			logger.Infoc(mCtx, "workerRunCallback: 总数[%d] worker均已执行结果[%s], callbackFunc 退出；", okCnt, result)
			return statList, result
		}
		select {
		case taskStat := <-statChan:
			statList = append(statList, *taskStat)
			okCnt++

			switch taskStat.Status {
			case Success:
				logger.Infoc(mCtx, "workerRunCallback: worker ID[%d] 状态[%s]已成功; 运行中[%d]", taskStat.ID, taskStat.Status, conCurrency-okCnt)
			case Fail:
				failCnt++
				logger.Errorc(mCtx, "workerRunCallback: worker ID[%d] 状态[%s]执行失败; 运行中[%d] 失败[%d]", taskStat.ID, taskStat.Status, conCurrency-okCnt, failCnt)
				errChan <- taskStat.err
				// 结束其他worker
				cancel()
			}
			logger.Infoc(mCtx, "workerRunCallback: worker 汇总[%d] 已完成[%d] 失败[%d] 运行中[%d]；执行结果：[%d][%s]-[%d][%d];", conCurrency, okCnt, failCnt, conCurrency-okCnt,
				taskStat.ID, taskStat.Status, taskStat.SuccessTask, taskStat.FailTask)
		case <-ctx.Done():
			// 结束其他worker
			logger.Errorc(mCtx, "workerRunCallback: 收到退出信号，结束等待任务流程: worker 汇总[%d] 已完成[%d] 失败[%d] 运行中[%d]；", conCurrency, okCnt, failCnt, conCurrency-okCnt)
			cancel()
		}
	}
	logger.Infoc(mCtx, "workerRunCallback: 开始接收 worker 执行结果,总计[%d] 已完成[%d]:", conCurrency, okCnt)

	return statList, result
}
